package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/4/30 10:34
 */
public class DWD_04_DwdTradeCardAddApp extends BaseSQLApp {
    public static void main(String[] args) {
        new DWD_04_DwdTradeCardAddApp().init("DwdTradeCardAddApp", 3005, 1, "DwdTradeCardAddApp");
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        // 1. 建表从kafka读取数据
        tEnv.executeSql("create table ods_db(" +
                            "   `database` string, " +
                            "   `table` string, " +
                            "   `type` string, " +
                            "   `ts` bigint, " +
                            "   `data` map<String, String>, " +
                            "   `old` map<String, String>," +
                            "   `pt` as proctime()  " +
                            ")" + SQLUtil.getKafkaDDL("DwdTradeCardAddApp", Constant.TOPIC_ODS_DB));
        
        
        // 2. 读取加购事实表数据
        Table cartAdd = tEnv
            .sqlQuery("select " +
                          "    data['id'] id, " +
                          "    data['user_id'] user_id, " +
                          "    data['sku_id'] sku_id, " +
                          "    data['source_id'] source_id, " +
                          "    data['source_type'] source_type, " +
                // 考虑到 购物车数据的变化, 新的数据 - 旧的数据
                // 只考虑数据新增的, 不考虑数据减少的
                          "    if(`type`='insert', cast(data['sku_num'] as int), cast(data['sku_num'] as int) - cast(`old`['sku_num'] as int)) sku_num, " +
                          "    ts, " +
                          "    pt " +
                          "from ods_db " +
                          "where `table`='cart_info' " +
                          "and ( " +
                          "   `type`='insert' " +
                          "    or ( " +
                          "        `type`='update' " +
                          "        and `old`['sku_num'] is not null " +
                          "        and  cast(data['sku_num'] as int) - cast(`old`['sku_num'] as int) > 0 " +
                          "    ) " +
                          ")");
        tEnv.createTemporaryView("cart_add", cartAdd);
        // 2. 读取 Mysql 中的字典表, 用户维度退化
        tEnv.executeSql(SQLUtil.getBaseDicLookUpDDL());
        // 3.  join两张表, 得到加购明细表
        Table result = tEnv.sqlQuery("select" +
                                        " cart_add.id, " +
                                        " user_id," +
                                        " sku_id," +
                                        " source_id, " +
                                        " dic_name as source_type_name," +
                                        " sku_num," +
                                        " ts " +
                                        "from cart_add " +
                                        "join base_dic for system_time as of cart_add.pt " +
                                        "on cart_add.source_type=base_dic.dic_code");
        
        // 4. 创建 uppsert 动态表
        tEnv.executeSql("create table dwd_trade_cart_add(" +
                            " id string, " +
                            " user_id string," +
                            " sku_id string," +
                            " source_id string, " +
                            " source_type_name string," +
                            " sku_num int," +
                            " ts bigint " +
                            ")" + SQLUtil.getKafkaDDL("DwdTradeCardAddApp", Constant.TOPIC_DWD_TRADE_CART_ADD));
        
        // 5. join 结果写入到 upsert 动态表
        result.executeInsert("dwd_trade_cart_add");
    }
}
/*
select
    data['id'] id,
    data['user_id'] user_id,
    data['sku_id'] sku_id,
    data['source_id'] source_id,
    data['source_type'] source_type,
    if(`type`='insert', data['sku_num'], cast(`data`['sku_num'] as int) - cast(old['sku_num'] as int) sku_num,
    ts,
    pt
from ods_db
where `table`='cart_info'
and (
   `type`='insert'
    or (
        `type`='update'
        and `old`['sku_num'] is no null
        and  cast(`data`['sku_num'] as int) - cast(old['sku_num'] as int) > 0
    )
)


 */